package org.finesys.common.redis.core.mq.stream;

import java.lang.reflect.Type;
import java.util.List;

import org.finesys.commom.jackson.util.JacksonUtil;
import org.finesys.common.redis.core.mq.client.RedisMQTemplate;
import org.finesys.common.redis.core.mq.interceptor.RedisMessageInterceptor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.stream.StreamListener;

import cn.hutool.core.util.TypeUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * Stream 消息监听器抽象类
 */

@Setter
@Getter
@Slf4j
public abstract class AbstractStreamMessageListener<T extends AbstractStreamMessage>
		implements StreamListener<String, ObjectRecord<String, String>> {

	/**
	 * 消息类型
	 */
	private final Class<T> messageType;

	/**
	 * key
	 */
	private final String streamKey;

	/**
	 * 消费者分组
	 */
	@Value("${spring.application.name}")
	private String group;

	private RedisMQTemplate redisMQTemplate;

	@SneakyThrows
	AbstractStreamMessageListener() {
		this.messageType = getMessageClass();
		this.streamKey = messageType.newInstance().getStreamKey();
	}

	private Class<T> getMessageClass() {
		Type type = TypeUtil.getTypeArgument(getClass(), 0);
		if (type == null) {
			throw new IllegalStateException(String.format("类型(%s)需要设置消息类型", getClass().getName()));
		}
		return (Class<T>) type;
	}

	@Override
	public void onMessage(ObjectRecord<String, String> message) {
		T messageObj = JacksonUtil.parseObject(message.getValue(), messageType);
		try {
			log.info("message:{}", messageObj);
			// 消费前处理
			consumeMessageBefore(messageObj);
			// 消费消息
			this.onMessage(messageObj);
			// ack 消息消费完成
			redisMQTemplate.getRedisTemplate().opsForStream().acknowledge(group, message);
		}
		finally {
			// 消费后处理
			consumeMessageAfter(messageObj);
		}

	}

	public abstract void onMessage(T message);

	private void consumeMessageAfter(T messageObj) {
		List<RedisMessageInterceptor> interceptors = redisMQTemplate.getInterceptors();
		for (int i = interceptors.size() - 1; i >= 0; i--) {
			interceptors.get(i).consumeMessageAfter(messageObj);
		}
	}

	private void consumeMessageBefore(T messageObj) {
		List<RedisMessageInterceptor> interceptors = redisMQTemplate.getInterceptors();
		interceptors.forEach(interceptor -> interceptor.consumeMessageBefore(messageObj));
	}

}
